package net.bwie.realtime.jtp.dwd;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import net.bwie.realtime.jtp.commp.utils.DateTimeUtil;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;

/**
 * @BelongsProject: realtime-project-10zlq
 * @BelongsPackage: net.bwie.realtime.jtp.dwd.log.function
 * @Author: zhangleqing
 * @CreateTime: 2025-08-16  13:49
 * @Description: TODO
 * @Version: 1.0
 */
public class AdjustIsNewProcessFunction extends KeyedProcessFunction<String, String, String> {

    // 声明状态
    private  transient ValueState<String> firstDateState = null;

    @Override
    public void open(Configuration parameters) throws Exception {
        // 初始化状态
        firstDateState = getRuntimeContext().getState(
            new ValueStateDescriptor<String>("first-date", String.class)
        );
    }


    @Override
    public void processElement(String value,
                               Context ctx,
                               Collector<String> out) throws Exception {
        // 获取状态值
        String firstDate = firstDateState.value();

        // 解析字段值
        JSONObject jsonObject = JSON.parseObject(value);
        String isNewValue = jsonObject.getJSONObject("common").getString("is_new");
        Long tsValue = jsonObject.getLong("ts");

        // 条件判断 如果是新用户
        if ("1".equals(isNewValue)){
            // 条件判断 如果首次日期未记录 直接赋值当前时间戳
            if (null == firstDate){
                firstDateState.update(DateTimeUtil.convertLongToString(tsValue, DateTimeUtil.DATE_FORMAT));
            }else {
                // 如果首次日期已记录 校验 “已记录的首次日期” 与 “当前时间戳对应的日期” 是否一致
                if (!firstDate.equals(DateTimeUtil.convertLongToString(tsValue, DateTimeUtil.DATE_FORMAT))){
                    jsonObject.getJSONObject("common").put("is_new", "0");
                }
            }
        }
        else {
            // 老用户如果首次日期未记录，则自动补全一个 “往前推一天” 的首次日期。
            if (null == firstDate){
                firstDateState.update(DateTimeUtil.convertLongToString(tsValue-864000000, DateTimeUtil.DATE_FORMAT));
            }
        }
        // 输出数据
        out.collect(jsonObject.toJSONString());
    }

}
